package com.example.kafkasp.Controller;

import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

/**
 * @author CHENTAIWEN
 * @create 2022/8/25 11:04
 */
public class KafkaProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
        props.put(ProducerConfig.RETRIES_CONFIG,0);
        props.put(ProducerConfig.LINGER_MS_CONFIG,1); //发送延迟 1秒
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554432);//管理的最大内存
        props.put(ProducerConfig.ACKS_CONFIG,1);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG,16384);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new org.apache.kafka.clients.producer.KafkaProducer<>(props);

        //3.在my-topic主题下，发送消息
        for(int i = 0; i < 10; i++) {
            producer.send(new ProducerRecord<>("Syan", Integer.toString(i), Integer.toString(i)), (metadata, exception) -> {
                if(exception == null) {
//                    在哪个分区，偏移量是多少
                    System.out.println("partition: " + metadata.partition() + " offset: " + metadata.offset());
                }
                else {
                    exception.printStackTrace();
                }
            });
        }
        //4.关闭
        producer.close();

    }
}
